package com.asiainfo.sqlextractor.adapter;

import com.alibaba.fastjson.JSONArray;
import com.asiainfo.sqlextractor.exec.DataSetHolder;
import com.asiainfo.sqlextractor.util.JavaTypeResolve;
import com.google.common.collect.Lists;
import org.apache.calcite.DataContext;
import org.apache.calcite.adapter.java.JavaTypeFactory;
import org.apache.calcite.linq4j.AbstractEnumerable;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.linq4j.Enumerator;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.util.Pair;

import java.util.List;

public class MemTable extends AbstractTable implements ScannableTable {

    private final TableDefine tableDefine;

    public MemTable(TableDefine tableDefine) {
        this.tableDefine = tableDefine;
    }

    @Override
    public Enumerable<Object[]> scan(DataContext root) {
        JSONArray data = DataSetHolder.get(tableDefine.name());
        return new AbstractEnumerable<Object[]>() {
            @Override
            public Enumerator<Object[]> enumerator() {
                return new MemEnumerator(tableDefine, data);
            }
        };
    }

    @Override
    public RelDataType getRowType(RelDataTypeFactory relDataTypeFactory) {
        JavaTypeFactory typeFactory = (JavaTypeFactory) relDataTypeFactory;
        List<String> names = Lists.newLinkedList();
        List<RelDataType> types = Lists.newLinkedList();
        JavaTypeResolve typeResolve = new JavaTypeResolve();
        for (Pair<String, Class<?>> nameType : tableDefine.columnTypes()) {
            RelDataType javaType = typeFactory.createJavaType(nameType.right);
            names.add(nameType.left);
            types.add(javaType);
        }
        return typeFactory.createStructType(Pair.zip(names, types));
    }

}
